package com.itcode;


import com.itcode.Deserializer.CustomerDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;

/**
 * Flink CDC的样例类
 *
 * @author code1997
 */
public class FlinkCdc {

    public static final String ENV = "env";
    public static final String LOCAL_ENV = "local";


    public static void main(String[] args) throws Exception {
        //注意远程提交的kerberos 认证问题
        String env = ParameterTool.fromArgs(args).get(ENV, LOCAL_ENV);
        if (LOCAL_ENV.equalsIgnoreCase(env)) {
            kerberosAuthentication();
        }
        //1.创建执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Flink-CDC将读取Binlog的位置信息以状态的方式保存在CK，如果想要做到断点续传，需要从Check point或者SavePoint上启动程序
        //2.1 开始checkpoint，5s一次
        executionEnvironment.enableCheckpointing(5000L);
        //2.2 指定CK的一致性语义
        executionEnvironment.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        executionEnvironment.getCheckpointConfig().setCheckpointTimeout(10000L);
        executionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        executionEnvironment.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
        //2.3 设置任务关闭的之前保存最后一次CK数据
        executionEnvironment.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 指定CK自动重启策略：设置值如果Flink job的重启策略，尝试3次，间隔2s. 默认情况下如何checkpoint那么就没有重启策略，如果设置了checkpoint，那么就会自动使用固定延迟重启策略。
        executionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
        //2.5 设置状态后端和checkpoint保存路径，主要用于故障恢复和断电续传：fs，Memory,Rocksdb
        //executionEnvironment.setStateBackend(new FsStateBackend("hdfs://hadoop02:8020/flinkCDC"));//过时的,使用HashMapStateBacked和Checkpoint来代替。。
        executionEnvironment.setStateBackend(new HashMapStateBackend());
        executionEnvironment.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop02:8020/flink/flink-cdc/check-point");
        //2.6 设置访问HDFS的用户名
        //我的hdfs环境设置了kerberos认证的方式，所以这里的代码可以去掉，如果是用户名代理的方式，则可以直接使用一下代码。
        //System.setProperty("HADOOP_USER_NAME", "code1997");
        //因为使用了kerberos认证，所以需要使用
        //3.创建Flink-mysql-cdc的source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop02")
                .port(3307)
                .username("root")
                .password("19971001")
                .databaseList("gmall_flink")
                //如果不指定会读取数据库下的所有表
                .tableList("gmall_flink.activity_info")
                //初始化读取，会读取全量数据，适合第一次连接.
                .startupOptions(StartupOptions.initial())
                //设置反序列化的方式
                .deserializer(new StringDebeziumDeserializationSchema())
                .build();
        //4.使用CDC source从mysql中读取数据
        DataStreamSource<String> mysqlDS = executionEnvironment.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
        //5.打印数据
        mysqlDS.print();
        //6.执行任务
        executionEnvironment.execute();

    }

    public static void kerberosAuthentication() {
        try {
            System.setProperty("java.security.krb5.conf", "src/main/resources/krb5/krb5.ini");
            UserGroupInformation.loginUserFromKeytab("code1997@CODE1997.COM", "src/main/resources/keytab/code1997.keytab");
            System.out.println(UserGroupInformation.getCurrentUser() + "=====" + UserGroupInformation.getLoginUser());
        } catch (IOException e) {
            throw new RuntimeException("认证失败");
        }

    }
}
